package ranks.tasks.readable;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;

import mapred.CombineConfig;
import mapred.InputConfig;
import mapred.Job;
import mapred.MapConfig;
import mapred.OutputConfig;
import mapred.ReduceConfig;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.KeyValueTextInputFormat;
import org.apache.hadoop.mapred.RunningJob;
import org.apache.hadoop.mapred.TextOutputFormat;

import ranks.base.Data;

public class Readable extends Job {

	private int lastRound;

	private String output;

	public Readable(int lastRound, String... settings) {
		super("Rank Readable", settings);
		this.lastRound = lastRound;
		this.output = this.getConfig("-output");
	}

	@Override
	protected CombineConfig getCombineConfig() {
		return CombineConfig.newInstance(null);
	}

	@Override
	protected InputConfig getInputConfig(FileSystem fs) throws IOException {
		return InputConfig.newInstance(KeyValueTextInputFormat.class, output
				+ "/Round_" + lastRound);
	}

	@Override
	protected MapConfig getMapConfig() {
		return MapConfig.newInstance(Readable_Mapper.class, 0, Text.class,
				Text.class, null);
	}

	@Override
	protected OutputConfig getOutputConfig(FileSystem fs) throws IOException {
		return OutputConfig.newInstance(TextOutputFormat.class, Text.class,
				Text.class, output + "/Final", null);
	}

	@Override
	protected ReduceConfig getReduceConfig() {
		return ReduceConfig.newInstance(null, 0);
	}

	@Override
	protected void setUp(JobConf conf, FileSystem fs) throws IOException {
		String z = "";
		try {
			z = new BufferedReader(new InputStreamReader(fs.open(new Path(
					output + "/Z_" + lastRound)))).readLine();
		} catch (Exception e) {
			z = new Data().toString();
		}
		this.setJobConfConfig("Z", z);
		
		long websize = Long.parseLong(new BufferedReader(new InputStreamReader(
				fs.open(new Path(output + "/WebSize")))).readLine());
		this.setJobConfConfig("WebSize", websize);
	}

	@Override
	protected void tearDown(JobConf conf, FileSystem fs, RunningJob job)
			throws IOException {
		System.exit(0);
	}

}
